package org.example.publisherrealtime.mapper.impl;

import lombok.extern.slf4j.Slf4j;
import org.apache.lucene.search.TotalHits;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.Operator;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.ParsedSum;
import org.elasticsearch.search.aggregations.metrics.SumAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.example.publisherrealtime.bean.NameValue;
import org.example.publisherrealtime.mapper.PublisherMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Repository;
import org.springframework.web.client.RestClientException;

import java.io.IOException;
import java.time.LocalDate;
import java.util.*;

@Slf4j
@Repository
public class PublisherMapperImpl implements PublisherMapper {

    @Autowired
    RestHighLevelClient esClient;

    private String orderPrefixName = "gmall_order_wide";


    @Override
    public Map<String, Object> searchDetailByItem(String itemName, String date, Integer from, Integer pageSize) {
        Map<String, Object> results = new HashMap<>();
        String indexName = orderPrefixName +date;
        SearchRequest searchRequest = new SearchRequest(indexName);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        MatchQueryBuilder matchQueryBuilder
                = QueryBuilders.matchQuery("sku_name", itemName).operator(Operator.AND);
        searchSourceBuilder.query(matchQueryBuilder);
        searchSourceBuilder.fetchSource(new String[]{"create_time","order_price","province_name","sku_name","sku_num"
                ,"total_amount","user_age","user_gender"},null);
        HighlightBuilder highlightBuilder = new HighlightBuilder().field("sku_name").preTags("<em>").postTags("</em>");
        searchSourceBuilder.highlighter(highlightBuilder);
        searchSourceBuilder.from(from);
        searchSourceBuilder.size(pageSize);
        searchRequest.source(searchSourceBuilder);
        try {
            SearchResponse searchResponse = esClient.search(searchRequest, RequestOptions.DEFAULT);
            long total = searchResponse.getHits().getTotalHits().value;
            SearchHit[] searchHits = searchResponse.getHits().getHits();
            ArrayList<Map> sourceMaps = new ArrayList<>();
            for(SearchHit searchHit : searchHits) {
                Map<String, Object> sourceAsMap = searchHit.getSourceAsMap();
                Map<String, HighlightField> highlightFields = searchHit.getHighlightFields();
                String skuNames = highlightFields.get("sku_name").fragments()[0].toString();
                sourceAsMap.put("sku_name", skuNames);
                sourceMaps.add(sourceAsMap);
            }
            results.put("total", total);
            results.put("detail", sourceMaps);
            return results;
        } catch (ElasticsearchStatusException ese) {
            if(ese.status() == RestStatus.NOT_FOUND) {
                log.warn(indexName + " not found");
            }
        }catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("搜索异常");
        }
        return results;
    }

    @Override
    public List<NameValue> searchStatByItem(String itemName, String date, String field) {
        ArrayList<NameValue> nameValues = new ArrayList<>();
        String indexName = orderPrefixName + date;
        SearchRequest searchRequest = new SearchRequest(indexName);

        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        MatchQueryBuilder matchQueryBuilder
                = QueryBuilders.matchQuery("sku_name", itemName).operator(Operator.AND);
        searchSourceBuilder.query(matchQueryBuilder);

        SumAggregationBuilder sumAggregationBuilder
                = AggregationBuilders.sum("total_amount").field("split_total_amount");
        TermsAggregationBuilder termsAggregationBuilder
                = AggregationBuilders.terms("groupby" + field).field(field).size(100);
        termsAggregationBuilder.subAggregation(sumAggregationBuilder);
        searchSourceBuilder.aggregation(termsAggregationBuilder);

        searchRequest.source(searchSourceBuilder);
        try {
            SearchResponse searchResponse = esClient.search(searchRequest, RequestOptions.DEFAULT);
            Aggregations aggregations = searchResponse.getAggregations();
            ParsedTerms parsedTerms = aggregations.get("groupby" + field);
            List<? extends Terms.Bucket> buckets = parsedTerms.getBuckets();
            for (Terms.Bucket bucket : buckets) {
                String key = bucket.getKeyAsString();
                Aggregations bucketAggregations = bucket.getAggregations();
                ParsedSum sumParsedTerms = bucketAggregations.get("total_amount");
                double value = sumParsedTerms.getValue();
                nameValues.add(new NameValue(key, value));
                return nameValues;

            }
        } catch (ElasticsearchStatusException ese) {
            if(ese.status() == RestStatus.NOT_FOUND) {
                log.warn(indexName + " not found");
            }
        }catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("搜索异常");
        }
        return nameValues;
    }



    @Override
    public Map<String, Object> searchDau(String td) {

        Long dauTotals = searchTotalDau(td);

        Map<String, Object> dauInfo = new HashMap<>();
        dauInfo.put("dauTotal",dauTotals);
        Map<String, Long> dauTd = searchDauHr(td);
        dauInfo.put("dauTd",dauTd);

        LocalDate Td = LocalDate.parse(td);
        LocalDate Yd = Td.minusDays(1);
        String yd = Yd.toString();

        Map<String, Long> dauYd = searchDauHr(yd);
        dauInfo.put("dauYd", dauYd);

        return dauInfo;
    }

    public Map<String, Long> searchDauHr(String td) {
        HashMap<String, Long> dauHr = new HashMap<>();
        String prefixName = "gmall_dau_info";
        String indexName = prefixName + td;
        SearchRequest searchRequest = new SearchRequest(indexName);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        //还是不需要详细数据
        searchSourceBuilder.size(0);
        //我们现在需要根据小时进行分组
        TermsAggregationBuilder termsAggregationBuilder
                = AggregationBuilders.terms("groupByHr").field("hr").size(24);
        searchSourceBuilder.aggregation(termsAggregationBuilder);
        searchRequest.source(searchSourceBuilder);
        //已将数据进行分组，现在执行语句
        try {
            SearchResponse searchResponse = esClient.search(searchRequest, RequestOptions.DEFAULT);
            Aggregations aggregations = searchResponse.getAggregations();
            //根据分组获取值 获取这种类型的就可以获取bucket，进行遍历ParsedTerms
            ParsedTerms parsedTerms = aggregations.get("groupByHr");
            List<? extends Terms.Bucket> buckets = parsedTerms.getBuckets();
            for (Terms.Bucket bucket : buckets){
                    //我们需要获取的是一个小时内的访问人数，所以我们先获取时间，在获取文档数量
                String hr = bucket.getKeyAsString();
                long hrTotal = bucket.getDocCount();
                dauHr.put(hr, hrTotal);
            }
            return dauHr;
        } catch (ElasticsearchStatusException ese) {
            if(ese.status() == RestStatus.NOT_FOUND) {
                log.warn(indexName + " not found");
            }
        }catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("搜索异常");
        }
        return dauHr;

    }

    @Override
    public Long searchTotalDau(String td){
        //这是从es里面获取总条数
        //创建索引前缀
        String prefixName = "gmall_dau_info";
        //拼接索引
        String indexName = prefixName + td;
        //创建search请求
        SearchRequest searchRequest = new SearchRequest(indexName);
        //编写查询规则，这里我们只需要获取总数
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.size(0);
        searchRequest.source(searchSourceBuilder);
        //执行查询，需要try catch 捕获异常
        try {
            SearchResponse searchResponse = esClient.search(searchRequest, RequestOptions.DEFAULT);
            Long dauTotals = searchResponse.getHits().getTotalHits().value;
            return dauTotals;
        } catch (ElasticsearchStatusException ese){
            //抛出找不到索引异常
            if(ese.status() == RestStatus.NO_CONTENT ){
                //这个打印日志的需要添加注解Slf4j
                log.warn(indexName + " not found");
                //没有找到直接返回0
                return 0L;
            }
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("查询失败...");
        }
        //只执行成功，返回总条数
        //执行失败，es的异常，抛出未找到索引
        //io异常，抛出异常，写数据
        //全部没成立，返回null
        return 0L;
    }

}
